package com.zhangdi.mgsync.core;

import com.mongodb.MongoChangeStreamException;
import com.mongodb.MongoCommandException;
import com.zhangdi.mgsync.core.adapter.MyDocument;
import com.zhangdi.mgsync.core.adapter.SyncFactory;
import com.zhangdi.mgsync.core.common.MgSyncConst;
import com.zhangdi.mgsync.core.listener.TopicMessageListener;
import org.apache.commons.lang3.StringUtils;
import org.bson.BsonDocument;
import org.springframework.data.mongodb.core.messaging.ChangeStreamRequest;
import org.springframework.data.mongodb.core.messaging.MessageListenerContainer;

import static org.springframework.data.mongodb.core.aggregation.Aggregation.match;
import static org.springframework.data.mongodb.core.aggregation.Aggregation.newAggregation;
import static org.springframework.data.mongodb.core.query.Criteria.where;


/**
 * Created by zhangdi on 2020/07/15 16:52:25.
 */
public class ChangeStreamListenerStartup extends TopicMessageListener {
    private MessageListenerContainer messageListenerContainer;

    public ChangeStreamListenerStartup(String scanAdapterPackage, SyncFactory<MyDocument> factory,
                                       MessageListenerContainer messageListenerContainer) {
        super(scanAdapterPackage, factory);
        this.messageListenerContainer = messageListenerContainer;
    }

    /**
     * ChangeStream从用户指定的数据token开始监听
     *
     * @throws MongoCommandException
     * @throws MongoChangeStreamException
     */
    public void startup() throws MongoCommandException, MongoChangeStreamException {
        String token = getCacheManager().get(MgSyncConst.KEY);
        register(token);
    }

    public void startup(String token) throws MongoCommandException, MongoChangeStreamException {
        register(token);
    }

    private void register(String token) {
        ChangeStreamRequest<Object> request = null;
        ChangeStreamRequest.ChangeStreamRequestBuilder<Object> builder = ChangeStreamRequest.builder(super::onMessage).
                filter(newAggregation(match(where("operationType").in("insert", "update", "delete"))));
        if (StringUtils.isBlank(token)) {
            request = builder.build();
        } else {
            BsonDocument resumeToken = BsonDocument.parse(token);
            request = builder.resumeToken(resumeToken).build();
        }
        messageListenerContainer.register(request);
    }
}
